package com.springcloudalibaba.baserocketmq.service.impl;

import com.springcloudalibaba.baserocketmq.service.RocketMqService;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.UUID;

/**
 * @描述: RocketMq消息服务 TODO 在发送消息的时候，我们只要把tags使用":"添加到topic后面就可以了
 * @作者: lixing
 * @日期 2021/6/8 21:07
 */
/* TODO 消费模板
@RocketMQMessageListener(
        topic = "", 				// topic：和消费者发送的topic相同
        consumerGroup = "",         // group：不用和生产者group相同
        selectorExpression = "*") 	// tag
public class MqListner implements RocketMQListener<T> {
    @Override
    public void onMessage(T t) {  }
}
*/
@Service
public class RocketMqServiceImpl<T> implements RocketMqService<T> {
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * @描述 发送普通消息
     * @作者 lixing
     * @日期 2021/6/8 21:07
     * @Param [topic, messageBody]
     */
    @Override
    public void sendSimpleMsg(String topic, T messageBody) {
        rocketMQTemplate.convertAndSend(topic, messageBody);
    }

    /**
     * @描述 发送【同步】【无序】消息
     * @作者 lixing
     * @日期 2021/6/8 21:11
     * @Param [topic, messageBody, hashKey, timeout]
     */
    @Override
    public SendResult sendSyncMsg(String topic, T messageBody, long timeout, int delayLevel) {
        SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(messageBody).build(), timeout, delayLevel);
        return sendResult;
    }

    /**
     * @描述 发送【同步】【有序】消息
     * @作者 lixing
     * @日期 2021/6/8 21:11
     * @Param [topic, messageBody, timeout]
     */
    @Override
    public SendResult sendSyncMsg(String topic, T messageBody, String hashKey, long timeout) {
        // 第三个参数是hashKey,会根据他的hash值计算发送到哪一个队列，我用的是同一个值,那么他们的hash一样就可以保证发送到同一个队列里
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, messageBody, hashKey, timeout);
        return sendResult;
    }

    /**
     * @描述 发送【异步】【无序】消息
     * @作者 lixing
     * @日期 2021/6/8 21:19
     * @Param [topic, messageBody, sendCallback, timeout, delayLevel]
     */
    @Override
    public void sendAsyncMsg(String topic, T messageBody, long timeout, int delayLevel) {
        rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(messageBody).build(),
                new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) { }
                    @Override
                    public void onException(Throwable throwable) { }
                }, timeout, delayLevel
        );
    }

    /**
     * @描述 发送【异步】【有序】消息
     * @作者 lixing
     * @日期 2021/6/8 21:19
     * @Param [topic, messageBody, sendCallback, timeout, delayLevel]
     */
    @Override
    public void sendAsyncMsgOrderly(String topic, T messageBody, String hashKey, long timeout) {
        // 第三个参数是hashKey,会根据他的hash值计算发送到哪一个队列，我用的是同一个值,那么他们的hash一样就可以保证发送到同一个队列里
        rocketMQTemplate.asyncSendOrderly(topic, messageBody, hashKey,
            new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) { }
                @Override
                public void onException(Throwable throwable) { }
        }, timeout);
    }

    /**
     * @描述 发送事务消息
     * @作者 lixing
     * @日期 2021/6/8 21:24
     * @Param [topic, tag, msg]
     * @return TransactionSendResult
     */
    @Override
    public TransactionSendResult sendMessageIntransaction(String topic, String tag, String msg) {
        Message<String> message = MessageBuilder.withPayload(msg)
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
                .setHeader(RocketMQHeaders.TAGS, "tag-lss0555")
                .setHeader("userid", "lss05555")
                .build();
        String destination = topic;
        TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, destination);
        return sendResult;
    }

}
